package com.lintf.distributed.transaction.inventory.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.lintf.distributed.transaction.DeductInventoryEvent;
import com.lintf.distributed.transaction.MessageProcessedEvent;
import com.lintf.distributed.transaction.inventory.entity.DeadLetterMessage;
import com.lintf.distributed.transaction.inventory.exception.BusinessException;
import com.lintf.distributed.transaction.inventory.exception.RetryStatisticsException;
import com.lintf.distributed.transaction.inventory.service.DeadLetterService;
import com.lintf.distributed.transaction.inventory.service.InventoryService;
import com.lintf.distributed.transaction.inventory.service.ProcessedMessageService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.aop.framework.AopContext;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.retry.ExhaustedRetryException;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.util.Collections;
import java.util.Date;
import java.util.List;

@Slf4j
@Component
@RequiredArgsConstructor
public class InventoryKafkaListener {

    private final InventoryService inventoryService;
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final RetryTemplate retryTemplate;
    private final ProcessedMessageService processedMessageService;
    private final DeadLetterService deadLetterService;

    /**
     * 处理库存扣减消息
     *
     * @param record 消息记录
     */
    @KafkaListener(topics = "deduct-inventory", errorHandler = "customKafkaErrorHandler")
    public void onMessage(ConsumerRecord<String, String> record) {
        String messageId = record.key();
        String json = record.value();

        log.info("收到库存扣减消息: key={}", messageId);

        try {
            // 1. 幂等性检查（在事务外）
            if (processedMessageService.isMessageProcessed(messageId)) {
                log.info("消息已处理过，跳过: key={}", messageId);
                return;
            }

            // 2. 消息解析（在重试外）
            DeductInventoryEvent event = parseEvent(json);

            InventoryKafkaListener proxy = (InventoryKafkaListener) AopContext.currentProxy();
            // 3. 使用重试机制执行业务逻辑
            retryTemplate.execute(context -> {
                // 在独立事务中执行业务
                return proxy.processInventoryDeduction(messageId, json, event);
            });
        } catch (Exception e) {
            // 其他异常处理
            handleProcessingError(record, json, e);
        }
    }

    /**
     * 处理消息处理错误
     *
     * @param record  消息记录
     * @param content 消息内容
     * @param e       异常信息
     */
    private void handleProcessingError(ConsumerRecord<String, String> record, String content, Exception e) {
        String messageId = record.key();
        String json = record.value();

        // 1. 获取根因异常和类型
        Throwable rootCause = getRootCause(e);
        String errorType = classifyError(e);

        // 2. 区分日志记录
        if (e instanceof ExhaustedRetryException) {
            log.error("重试耗尽[原始原因: {}]: MessageId[{}]，ErrorMsg[{}]", rootCause.getMessage(), messageId, e.getMessage());
        } else {
            log.error("处理失败[{}]: MessageId[{}]，ErrorMsg[{}]", errorType, messageId, e.getMessage());
        }
        log.error("库存扣减失败[{}]: MessageId[{}]", errorType, messageId);

        // 提取重试统计信息
        int actualRetries = 0;
        List<Long> retryTimes = Collections.emptyList();

        // 查找附加的统计异常
        for (Throwable suppressed : e.getSuppressed()) {
            if (suppressed instanceof RetryStatisticsException stats) {
                actualRetries = stats.getRetryCount();
                retryTimes = stats.getRetryTimes();
                break;
            }
        }

        log.error("重试统计: 尝试次数={}, 重试时间={}，重试耗时={}s", actualRetries, retryTimes, (retryTimes.get(retryTimes.size() - 1) - retryTimes.get(0)) / 1000);

        // 保存死信记录
        DeadLetterMessage deadLetter = deadLetterService.saveDeadLetter(
                messageId,
                json,
                record.topic(),
                e,
                actualRetries
        );

        // 异步处理后续操作
        kafkaTemplate.send("deduct-inventory-dlt", messageId, JSON.toJSONString(deadLetter));

        // 异步发送处理结果事件
        sendMessageProcessedEventAsync(deadLetter.getMessageKey(), content, false, errorType);

        // 异步标记消息已处理（防止重复处理）
        processedMessageService.markMessageProcessed(deadLetter.getMessageKey());
    }

    /**
     * 获取根因异常
     */
    private Throwable getRootCause(Throwable e) {
        Throwable cause = e;
        while (cause.getCause() != null && cause.getCause() != cause) {
            cause = cause.getCause();
        }
        return cause;
    }

    /**
     * 错误分类
     */
    private String classifyError(Exception e) {
        if (e instanceof JSONException) {
            return "消息格式错误";
        } else if (e instanceof ExhaustedRetryException) {
            return "重试耗尽";
        } else if (e instanceof BusinessException) {
            return "业务错误";
        }
        return "系统错误";
    }

    /**
     * 解析扣减库存事件
     *
     * @param json 消息内容
     * @return DeductInventoryEvent
     * @throws JSONException 如果解析失败
     */
    private DeductInventoryEvent parseEvent(String json) {
        try {
            return JSON.parseObject(json, DeductInventoryEvent.class);
        } catch (JSONException e) {
            log.error("消息解析失败: {}", json, e);
            throw e;
        }
    }

    /**
     * 处理库存扣减逻辑
     *
     * @param messageId 消息ID
     * @param json      消息内容
     * @param event     扣减库存事件
     * @return Void
     */
    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
    public Void processInventoryDeduction(String messageId, String json, DeductInventoryEvent event) {
        // 1. 执行业务逻辑
        boolean success = inventoryService.deductStock(event.getProductId(), event.getQuantity());

        if (!success) {
            log.warn("库存扣减失败: product={}, quantity={}",
                    event.getProductId(), event.getQuantity());
            throw new BusinessException("库存扣减失败");
        }

        // 2. 标记消息已处理（与业务操作在同一事务）
        processedMessageService.markMessageProcessed(messageId);

        // 3. 发送处理结果事件（异步，不阻塞事务）
        sendMessageProcessedEventAsync(messageId, json, true, null);

        log.info("库存扣减成功: product={}, quantity={}",
                event.getProductId(), event.getQuantity());
        return null;
    }

    /**
     * 异步发送消息处理结果事件
     *
     * @param messageId 消息ID
     * @param content   消息内容
     * @param success   是否成功
     * @param reason    失败原因
     */
    private void sendMessageProcessedEventAsync(String messageId, String content, boolean success, String reason) {
        try {
            MessageProcessedEvent event = new MessageProcessedEvent(
                    messageId,
                    success,
                    content,
                    reason,
                    new Date()
            );

            kafkaTemplate.send("message-processed", messageId, JSON.toJSONString(event));
            log.info("发送消息处理结果: key={}, success={}, reason={}",
                    messageId, success, reason);
        } catch (Exception e) {
            log.error("发送处理结果事件失败: key={}", messageId, e);
        }
    }

}